package com.shopping.hotItem;

import com.shopping.hotItem.bean.Result;
import com.shopping.hotItem.bean.UserBehavior;
import com.shopping.hotItem.bean.WindResult;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.shopping.constant.DataPath;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * 需求：每隔五分钟输出最近一个小时内点击量最多的前n个商品
 */
public class HotItemMain {
    public static void main(String[] args)  throws Exception{
        //创建环境，设置参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> source = env.readTextFile(DataPath.USER_BEHAVIOR);
        SingleOutputStreamOperator<UserBehavior> userBehavior = source.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String s) throws Exception {
                try {
                    String[] split = s.split(",");
                    return new UserBehavior(Long.parseLong(split[0]),
                            Long.parseLong(split[1]),
                            Integer.parseInt(split[2]),
                            split[3],
                            Long.parseLong(split[4]));
                } catch (Exception e) {
                    return null;
                }
            }
        })
                // 使用事件时间
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
                    @Override
                    public long extractAscendingTimestamp(UserBehavior userBehavior) {
                        return userBehavior.getTimeStamp() * 1000L;
                    }
                });

        KeyedStream<UserBehavior, Long> keyedStream = userBehavior.keyBy(UserBehavior::getItemId);
        WindowedStream<UserBehavior, Long, TimeWindow> windowedStream = keyedStream.timeWindow( Time.hours(1),Time.minutes(5));

        // 开窗进行聚合
        SingleOutputStreamOperator<WindResult> aggregate = windowedStream.aggregate(new AggregateFunction<UserBehavior, Integer, Integer>() {
            @Override
            public Integer createAccumulator() {
                return 0;
            }

            @Override
            public Integer add(UserBehavior a, Integer integer) {
                return integer + 1;
            }

            @Override
            public Integer getResult(Integer integer) {
                return integer;
            }

            @Override
            public Integer merge(Integer integer, Integer acc1) {
                return integer + acc1;
            }
        }, new WindowFunction<Integer, WindResult, Long, TimeWindow>() {
            @Override
            public void apply(Long aLong, TimeWindow timeWindow, Iterable<Integer> iterable, Collector<WindResult> collector) throws Exception {
                // 第一个参数是keyBy的kay
                Integer count = iterable.iterator().next();
                collector.collect(new WindResult(aLong, count,timeWindow.getStart()));
            }
        });
        aggregate.print();

        // 根据同一时间窗口进行开窗求取topn
        aggregate.keyBy(WindResult::getWindow)
                .process(new KeyedProcessFunction<Long, WindResult, Object>() {
                    @Override
                    public void processElement(WindResult windResult, Context context, Collector<Object> collector) throws Exception {

                    }
                });


        //启动执行
        env.execute("HotItem");

    }
}
